package com.tyhy.program.cbc.mobileapi.redis.websocket;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * @ServerEndpoint(value="/websocket")value值必须以/开路 备注:@ServerEndpoint注解类不支持使用@Autowire
 * {topic}指：向哪个频道主题里发消息
 * {myname}指：这个消息是谁的。真实环境里可以使用当前登录用户信息
 */
@Slf4j
@Component
@ServerEndpoint(value = "/websocket/{topic}")
@Data
public class WebsocketEndpoint {

    /**
     * 因为@ServerEndpoint不支持注入，所以使用SpringUtils获取IOC实例
     */
    private StringRedisTemplate redisTampate = SpringUtils.getBean(StringRedisTemplate.class);

    private RedisMessageListenerContainer redisMessageListenerContainer = SpringUtils.getBean(RedisMessageListenerContainer.class);

    //存放该服务器该ws的所有连接。用处：比如向所有连接该ws的用户发送通知消息。
    private static CopyOnWriteArraySet<WebsocketEndpoint> sessions = new CopyOnWriteArraySet<>();

    private Session session;

    /**
     * 临时存储 SubscribeListener
     */
    private static ConcurrentHashMap<String, SubscribeListener> SUBSCRIBE_LISTENER_MAP = new ConcurrentHashMap<>();
    /**
     * 临时存储 CHANNEL_TOPIC_MAP
     */
    private static ConcurrentHashMap<String, ChannelTopic> CHANNEL_TOPIC_MAP = new ConcurrentHashMap<>();

    @OnOpen
    public void onOpen(Session session, @PathParam("topic") String topic) {
        log.info("topic:{},java websocket:打开连接", topic);

        //如果已经打开过,先进行删除处理
        SubscribeListener oldSubscribeListener = SUBSCRIBE_LISTENER_MAP.get(topic);
        if (oldSubscribeListener != null) {
            log.info("删除重复订阅,topic={}", topic);
            redisMessageListenerContainer.removeMessageListener(oldSubscribeListener);
        }


        //临时存储 对象
        this.session = session;
        sessions.add(this);
        SubscribeListener subscribeListener = new SubscribeListener();
        subscribeListener.setSession(session);
        subscribeListener.setStringRedisTemplate(redisTampate);

        ChannelTopic channelTopic = new ChannelTopic(topic);
        CHANNEL_TOPIC_MAP.put(topic, channelTopic);
        SUBSCRIBE_LISTENER_MAP.put(topic, subscribeListener);

        //设置订阅topic


        redisMessageListenerContainer.addMessageListener(subscribeListener, new ChannelTopic(topic));
        log.info("topic:{},当前listener总数:{},redis 订阅 topic,添加 listener", topic, SUBSCRIBE_LISTENER_MAP.size());
    }

    @OnClose
    public void onClose(Session session, @PathParam("topic") String topic) {
        log.info("topic:{},java websocket:关闭连接", topic);

        //关闭连接,释放内存对象
        sessions.remove(this);
        ChannelTopic channelTopic = CHANNEL_TOPIC_MAP.get(topic);
        SubscribeListener subscribeListener = SUBSCRIBE_LISTENER_MAP.get(topic);

        CHANNEL_TOPIC_MAP.remove(topic);
        SUBSCRIBE_LISTENER_MAP.remove(topic);
        log.info("redisMessageListenerContainer删除topic,listener开始,redisMessageListenerContainer:{},topic:{},listener:{}", redisMessageListenerContainer, channelTopic, subscribeListener);
        if (null != redisMessageListenerContainer && null != subscribeListener && null != channelTopic) {
            try {
                redisMessageListenerContainer.removeMessageListener(subscribeListener, channelTopic);
                log.info("redisMessageListenerContainer删除topic,listener结束,redisMessageListenerContainer:{},topic:{},listener:{}", redisMessageListenerContainer, CHANNEL_TOPIC_MAP.get(topic), SUBSCRIBE_LISTENER_MAP.get(topic));
            } catch (Exception e) {
                log.info("topic:{},redis 删除异常(已处理),当前listener总数:{},redis 删除 topic,remove listener", topic, SUBSCRIBE_LISTENER_MAP.size());
            }
        }

        log.info("topic:{},当前listener总数:{},redis 删除 topic,remove listener", topic, SUBSCRIBE_LISTENER_MAP.size());
    }
    @OnMessage
    public void onMessage(Session session, String message, @PathParam("topic") String topic) throws IOException {
        log.info("topic:{},java websocket 收到消息=={}", topic, message);
        message = topic + "：" + message;
        PublishService publishService = SpringUtils.getBean(PublishService.class);
        publishService.publish(topic, message);
        log.info("topic:{},java websocket 消息已发送=={}", topic, message);
    }

//    @OnMessage
//    public void onMessage(Session session, String message, @PathParam("topic") String topic, @PathParam("myname") String myname) throws IOException {
//        log.info("topic:{},java websocket 收到消息=={}", topic, message);
//        message = myname + "：" + message;
//        PublishService publishService = SpringUtils.getBean(PublishService.class);
//        publishService.publish(topic, message);
//        log.info("topic:{},java websocket 消息已发送=={}", topic, message);
//    }

    @OnError
    public void onError(Session session, Throwable error) {
        log.error("java websocket 出现错误", error);
    }

}